package trunk.third;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;

import java.io.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;

public class daMapperHash extends MapReduceBase implements Mapper<IntWritable, BytesWritable, Text, IntWritable>{
    private static HashMap<Byte, ArrayList<byte[]>> hashMap = new HashMap<>();
    private int min = Integer.MAX_VALUE;

    public void configure(JobConf job){

        hashMap.put((byte)97 , new ArrayList<byte[]>());
        hashMap.put((byte)98 , new ArrayList<byte[]>());
        hashMap.put((byte)99 , new ArrayList<byte[]>());
        hashMap.put((byte)100, new ArrayList<byte[]>());

        try {
            BufferedReader reader = new BufferedReader(new InputStreamReader( new FileInputStream(new File(DistributedCache.getCacheFiles(job)[0].getPath()))));
            String line;
            while((line = reader.readLine())!=null)
            {
                if (line.length() - 1 < min) min = line.length() - 1;
                hashMap.get((byte)line.charAt(0)).add(line.getBytes());
            }


        } catch (IOException e){ e.printStackTrace(); }
    }
    @Override
    public void map(IntWritable keyIn, BytesWritable valueIn, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
        byte[] input = valueIn.getBytes();  int blockStart = keyIn.get();
        Text keyOut = new Text();           IntWritable valueOut = new IntWritable();   int patternLength;
        String currentFile = ((FileSplit)reporter.getInputSplit()).getPath().getName();
        Iterator<byte[]> iterator;

        output.collect(keyOut, new IntWritable(-1)); // try make every query appear
        for (int runner = 0; runner != blockStart + input.length - min - 1; runner++){
            iterator = hashMap.get(input[runner]).iterator();
            while (iterator.hasNext()){
                byte[] current = iterator.next();
                for (int i = current.length - 1; i != 0; i--){
                    if (input[runner + i] == current[i])
                    {
                        if (i == 1){
                            keyOut.set(currentFile + "," + new String(current)); // set key
                            valueOut.set(keyIn.get() + runner); // set value
                            output.collect(keyOut, valueOut);
                        }
                        else
                            continue;
                    }
                    else
                        break;
                }
            }



        }
    }
}
